package org.apache.kylin.engine.mr.steps;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.kylin.common.util.ILock;
import org.apache.kylin.engine.mr.HadoopUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 * Created by taoliu on 2016/11/25.
 */
public class TaskHdfsLock implements ILock {
    private static final Logger logger = LoggerFactory.getLogger(TaskHdfsLock.class);
    private static String lockRootPath;
    private static Object execLock = new Object();
    public TaskHdfsLock(String lockRoot) {
        if (StringUtils.isEmpty(lockRootPath)){
//            String lockRoot = KylinConfig.getInstanceFromEnv().getMergeTaskLockRoot();
            lockRootPath = lockRoot.endsWith("/") ? lockRoot : (lockRoot + "/");
        }
    }


    @Override
    public boolean lock(String cubeName) throws Exception {
        if (StringUtils.isEmpty(cubeName)){
            throw new Exception("cubeName could not be null");
        }
        synchronized (execLock){
            try {
                String lockName = cubeName + ".lock";
                String lockPath = lockRootPath + lockName;
                Path path = new Path(lockPath);
                if (HadoopUtil.getFileSystem().exists(path)){
                    logger.info("lock is exist,will not Add new merge task lock");
                    return false;
                }
                HadoopUtil.getFileSystem().mkdirs(path);
                logger.info(" Add merge task lock success ");
            } catch (IOException e) {
                logger.error("Add merge task lock err",e);
                throw e;
            }
            return true;
        }
    }

    @Override
    public boolean lockExist(String cubeName) throws Exception {
        try {
            Path path = getLockPath(cubeName);
            return HadoopUtil.getFileSystem().exists(path);
        } catch (IOException e) {
            logger.error("lock exist err",e);
            throw e;
        }
    }

    @Override
    public boolean unlock(String cubeName) throws IOException {
        try {
            Path path = getLockPath(cubeName);
            if (HadoopUtil.getFileSystem().exists(path)){
                HadoopUtil.getFileSystem().delete(path,false);
                logger.info(" Remove merge task lock success");
                return true;
            }
            logger.info(" no merge task lock need to remove ");
        } catch (IOException e) {
            logger.error("Remove merge task lock err",e);
            throw e;
        }
        return false;
    }

    private Path getLockPath(String cubeName) throws IOException {
        String lockName = cubeName + ".lock";
        String lockPath = lockRootPath + lockName;
        return new Path(lockPath);
    }
}
